package pubsub.impl;

import client.DtClientAware;
import pubsub.MessageListener;
import pubsub.Subscriber;

import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Created by xuan on 2018/8/16.
 */
public class DefaultSubscriber extends DtClientAware implements Subscriber {

    private Map<String, MessageListener> listenerMap = new ConcurrentHashMap<>();

    @Override
    public void addListener(String topic, MessageListener listener) {
        listenerMap.put(topic, listener);
    }

    @Override
    public void doSubscribe() {
        Set<String> topicSet = listenerMap.keySet();
        String[] topics = topicSet.toArray(new String[topicSet.size()]);

        new Thread(() -> getClient().subcribe((String topic, String message) -> {
            MessageListener l = listenerMap.get(topic);
            if (null != l) {
                l.onMessage(topic, message);
            }
        }, topics)).start();
    }

}
